package pku;



import pku.consumer.BufferService;
import pku.consumer.MessageReader;
import pku.serializer.MessageSerializer;

import java.util.*;

public class Consumer {
   // private static final Logger logger = LoggerFactory.getLogger(Consumer.class);

    private KeyValue properties;
    private BufferService bufferService;
    private ArrayList<MessageReader> readers = new ArrayList<>();
    private int pollIndex = 0;
    private int count = 0;

    private MessageSerializer deserializer = new MessageSerializer(); // thread local
    
    public Consumer(KeyValue properties) {
        this.properties = properties;
        this.bufferService = BufferService.getInstance(properties.getString("STORE_PATH"));
    }

    public Consumer() {
        this.bufferService = BufferService.getInstance("data");
    }

     public KeyValue properties() {
        return properties;
    }

     public synchronized ByteMessage poll() {
        MessageReader reader = readers.get(pollIndex);
         ByteMessage message = reader.readMessage();
        while (message == null) {
            readers.remove(pollIndex);
            if (readers.isEmpty()) return null;
            pollIndex = pollIndex % readers.size();
            reader = readers.get(pollIndex);
            message = reader.readMessage();
        }
        if ((++count & 0x3f) == 0) { // change buffer every 64 messages
            pollIndex = (pollIndex + 1) % readers.size();
            
            if (Constants.ENABLE_MESSAGE_SAMPLING) {
                // FOR DEBUG: Sample and print message every 2^17 (~131K) messages
                if ((count & 0x1ffff) == 0) {
                    //logger.info("Sampled message: {}", message.toString());
                }
            }
        }

         Iterator<Map.Entry<String, Object>> iterator=message.headers().getMap().entrySet().iterator();//创建iterator对象
         while(iterator.hasNext()){//用while循环，判断是否有下一个

             Map.Entry<String, Object> entry=iterator.next();//声明entry，并用它来装载字符串
             System.out.println(new Date() + " " + message.headers().getMap().size() + " " + " key="+entry.getKey()+" valve="+entry.getValue() + " " + Thread.currentThread().getName()
              + " " + message.getBody());//调用getKey()和getValue()打印
         }

        return message;
    }

     public ByteMessage poll(KeyValue properties) {
        throw new UnsupportedOperationException("Unsupported");
    }

     public void ack(String messageId) {
        throw new UnsupportedOperationException("Unsupported");
    }

     public void ack(String messageId, KeyValue properties) {
        throw new UnsupportedOperationException("Unsupported");
    }

     public synchronized void attachQueue(String queueName, Collection<String> topics) {
        readers.add(new MessageReader(queueName, true, bufferService, deserializer));
        for (String topic: topics) {
            readers.add(new MessageReader(topic, false, bufferService, deserializer));
        }
    }
}
